#include "common.h"
#include "cuckoo.h" //cuckoo hashmap
#include "arraylist.h"

//======================global var====================
typedef struct NodeMeta
{
	int nodeID;
	int worker2master_sockfd[ClientNum];//master作为服务端接受worker的connect
	long blockNum;
	long requestNum;
	/*
	*some other attrs 
	*
	*/
}NodeMeta;

typedef struct RackMeta
{
	int rackID;
	long blockNum;
	long requestNum;
	/*
	*some other attrs 
	*
	*/
}RackMeta;

pthread_rwlock_t lock4EnvMeta;

typedef struct MasterAsClient2WorkerConnect
{
	int nodeID;
	int master2worker_sockfd;//master作为客户端向worker发出connect
}MasterAsClient2WorkerConnect;

cuckoo* blockID2MetaMap=NULL;//blockID与blockMeta之间的hashmap
pthread_rwlock_t lock4BlockMeta;

cuckoo* blockID2SecondMap=NULL;//主从副本BLock之间的hashmap


cuckoo* RGID2RGroupMap=NULL;//条带ID与条带信息之间的hashmap
cuckoo* parityID2MetaMap=NULL;//校验数据块与blockMeta之间的hashmap

Node* nodes=NULL;//节点信息
Rack* racks=NULL;//机架信息
int nodeID=-1;//当前节点的ID
int isDestory=0;
long RGID=0;//RG_ID从0开始递增 也表示当前总共构建的条带数目
int sort_way;

int currentRackID;//当前的primary rackID
int setCount;//当前一组set的count 达到K时置为0
int rack4SecondaryBlock[RackNumTotal];//当前主rack的另一个rack_index
pthread_rwlock_t lock4SelectingRack;

int client2master_sockfd[ClientNum]={-1};
//int client2worker_sockfd[NodeNumTotal-1];//client作为客户端与worker之间的通信
MasterAsClient2WorkerConnect* masterAsClient2WorkerConnect;
NodeMeta* nodeMetas=NULL;
RackMeta* rackMetas=NULL;
Array_List* rackList[RackNumTotal]={NULL};//存储每个rack中unpopular的数据块
Array_List* unpopularBlockList=NULL;
Array_List* ready2EncodingRGList=NULL;

Array_List* blockList=NULL;//存储所有未编码的block
pthread_rwlock_t lock4BlockList;
FILE *archival_file=NULL;
FILE *load_file=NULL;
int isBegin_archival_Record=0;
//====================================================

//====================function========================
int init_lock4master();
int init_master();  //先初始化master
int selectRack();
int selectNode(int nodeID);
int determineUnpopularBlocks();
int constructStripes(int flag);
void sortByLocality(int flag);
int constructStripesInRack(int rackID);
void processArchival();
int selectEncodingNode(ArchivalRequest* archivalRequest,int rackID);
void master2worker_send(ArchivalRequest *request,int nodeID);
void destory_master();
void destory_lock4master();
void* archival_thread(void* arg);
void printCondition();

int handlerRequestIOFromClient(int client2master_fd);
int client2master_recv(Client2MasterRequest *request,int sockfd);
void master2client_send_Resp4Read(Master2ClientResponse4Read* master2ClientResponse4Read,int sockfd);
void master2client_send_Resp4Write(Master2ClientResponse4Write* master2ClientResponse4Write,int clientID);
int handlerWriteRequestFromClient(long long blockID,long requestID,int clientID);
int handlerReadRequestFromClient(long long blockID,long requestID,int clientID);


int handlerRequestIOFromWorker(int worker2master_fd);
int worker2master_recv(Worker2MasterRequest* request,int sockfd);

void* process4worker_thread(void* arg);
void* process4client_thread(void* arg);
//====================================================
int init_lock4master(){
	if (pthread_rwlock_init(&lock4BlockMeta,NULL)<0) return ERR;
	if (pthread_rwlock_init(&lock4EnvMeta,NULL)<0) return ERR;
	if (pthread_rwlock_init(&lock4SelectingRack,NULL)<0) return ERR;
	if (pthread_rwlock_init(&lock4BlockList,NULL)<0) return ERR;
	return 0;
}

int init_master(){
	printf("******************************init_master begin ................\n");
	srand((int)time(NULL));//设置随机函数的seed
	if (init_lock4master()<0)
	{
		printf("init_master()====init_lock4master error\n");
		return ERR;
	}

	if (set_server_limit() <0) {
      	printf("init_master()====set_server_limit error\n");
		return ERR;
    }
	//1.初始化变量
	int i;
	currentRackID=0;
	setCount=0;
	for (i = 0; i < RackNumTotal; ++i)
	{
		rack4SecondaryBlock[i]=(i+1)%RackNumTotal;
		rackList[i] = Array_List_Init();
	}
	sort_way=Sort_Way;
	blockList = Array_List_Init();
	unpopularBlockList=Array_List_Init();
	ready2EncodingRGList=Array_List_Init();
    blockID2MetaMap = cuckoo_init(100);
    blockID2SecondMap = cuckoo_init(100);
    RGID2RGroupMap = cuckoo_init(50);
    parityID2MetaMap=cuckoo_init(50);
    printf("init_master hashmap init end............................\n");
	//2.初始化配置文件
	nodes=read_config_Node(CONFIG_NODE);
	printf("read_config_Node end............................\n");
	racks=read_config_Rack(CONFIG_RACK);
	printf("read_config_Rack end............................\n");

	nodeID = current_node(nodes);
	if(nodeID==-1) {
    	printf("node_init()====current_node is error\n");
    	return ERR;  
    }
	printf("current_node:%d............................\n",nodeID);

	//3.初始化master的网络通信
	
	// for (int i = 0; i < NodeNumTotal-1; ++i)
	// {
	// 	client2worker_sockfd[i]=connect_try(nodes[i].ip,Client2WorkerPort);
	// 	if (client2worker_sockfd[i]!=-1)
	// 	{
	// 		printf("client2worker_sockfd successed and client2worker_sockfd[%d]=%d \n",i,client2worker_sockfd[i]);
	// 	}else{
	// 		printf("client2worker_sockfd is error and client2worker_sockfd[%d]=%d \n",i,client2worker_sockfd[i]);
 //     		return ERR;
	// 	}
	// }

	//3.2初始化master作为客户端尝试连接worker服务器
	masterAsClient2WorkerConnect = (MasterAsClient2WorkerConnect*)malloc(sizeof(MasterAsClient2WorkerConnect)*(NodeNumTotal-1));
	for (int i = 0; i < NodeNumTotal-1; ++i)
	{
		int master2worker_sockfd;
		master2worker_sockfd = connect_try(nodes[i].ip,Master2WorkerPort);
		if (master2worker_sockfd!=-1)
		{
			printf("master2worker_sockfd successed and master2worker_sockfd[%d]=%d \n",i,master2worker_sockfd);
			masterAsClient2WorkerConnect[i].nodeID=nodes[i].nodeID;
			masterAsClient2WorkerConnect[i].master2worker_sockfd=master2worker_sockfd;
		}else{
			printf("master2worker_sockfd is error and master2worker_sockfd[%d]=%d \n",i,master2worker_sockfd);
     		return ERR;
		}
	}

	//3.3初始化master作为服务器端与各个worker之间的连接 注意端口需要加上nodeID的值
	nodeMetas = (NodeMeta*)malloc(sizeof(NodeMeta)*(NodeNumTotal-1));
	for (int i = 0; i < NodeNumTotal-1; ++i)
	{
		for (int j = 0; j < ClientNum; ++j)
		{
			int worker2master_sockfd;
			worker2master_sockfd=server_accept(Worker2MasterPort+i,Max_Length_backlog);
			if (worker2master_sockfd!=-1)
			{
				printf("worker2master_sockfd successed and worker2master_sockfd[%d]=%d \n",i,worker2master_sockfd);
				nodeMetas[i].nodeID=i;
				nodeMetas[i].worker2master_sockfd[j]=worker2master_sockfd;
			}else{
				printf("worker2master_sockfd is error and worker2master_sockfd[%d]=%d \n",i,worker2master_sockfd);
	     		return ERR;
			}
		}
	}



	//rack元信息初始化
	rackMetas = (RackMeta*)malloc(sizeof(RackMeta)*RackNumTotal);
	for (i = 0; i < RackNumTotal; ++i)
	{
		rackMetas[i].rackID=i;
		rackMetas[i].blockNum=0;
		rackMetas[i].requestNum=0;
	}

	//node其他元信息初始化
	for (i = 0; i < NodeNumTotal-1; ++i)
	{
		nodeMetas[i].requestNum=0;
		nodeMetas[i].blockNum=0;
	}

	//master作为服务端等待client的连接
	for (int i = 0; i < ClientNum; ++i)
	{
		client2master_sockfd[i]=server_accept(Client2MasterPort,Max_Length_backlog);
	 	if (client2master_sockfd[i]!=-1)
	 	{
	 		printf("node_init()====node accept client  successed\n"); 
	 	}else{
	 		printf("node_init()====node accept client error\n");
        	return ERR;
	 	}
	}


	// traces = (Trace*)malloc(sizeof(Trace)*Max_Trace);
	// traceNum = init_trace(traces,TRACE);

	printf("init_master end===========================\n");
  	return 0;
}

void destory_lock4master(){
	pthread_rwlock_destroy(&lock4BlockMeta);
	pthread_rwlock_destroy(&lock4EnvMeta);
	pthread_rwlock_destroy(&lock4SelectingRack);
	pthread_rwlock_destroy(&lock4BlockList);
}

//销毁master资源
void destory_master(){
	destory_lock4master();
	if (nodes!=NULL) free(nodes);
	if (racks!=NULL) free(racks);
	//if (traces!=NULL) free(traces);
	if (blockID2MetaMap!=NULL) cuckoo_destroy(blockID2MetaMap,1);
	if (blockID2SecondMap!=NULL) cuckoo_destroy(blockID2SecondMap,1);
	if (parityID2MetaMap!=NULL) cuckoo_destroy(parityID2MetaMap,1);
	if (RGID2RGroupMap!=NULL) cuckoo_destroy(RGID2RGroupMap,1);
	for (int i = 0; i < RackNumTotal; ++i)
	{
		if (rackList[i]!=NULL) Array_List_Free(rackList[i]);
	}
	if (blockList!=NULL) Array_List_Free(blockList);
	if (unpopularBlockList!=NULL) Array_List_Free(unpopularBlockList);
	if (ready2EncodingRGList!=NULL) Array_List_Free(ready2EncodingRGList);

	if (nodeMetas!=NULL){
		for (int i = 0; i < NodeNumTotal-1; ++i){
				for (int j = 0; j < ClientNum; ++j)
				{
					if (nodeMetas[i].worker2master_sockfd[j]!=-1) close(nodeMetas[i].worker2master_sockfd[j]);
				}
		}
		free(nodeMetas);
	}
	if (rackMetas!=NULL) free(rackMetas);

	if (masterAsClient2WorkerConnect!=NULL)
	{
		for (int i = 0; i < NodeNumTotal-1; ++i){
			if (masterAsClient2WorkerConnect[i].master2worker_sockfd!=-1) close(masterAsClient2WorkerConnect[i].master2worker_sockfd);
		}
		free(masterAsClient2WorkerConnect);
	}

	// for (int i = 0; i < NodeNumTotal-1; ++i){
	// 	if (client2worker_sockfd[i]!=-1) close(client2worker_sockfd[i]);
	// }

	for (int i = 0; i < ClientNum; ++i)
	{
		if (client2master_sockfd[i]!=-1) close(client2master_sockfd[i]);
	}

}

void* process4worker_thread(void* arg){
	int sockfd=(int)arg;
	int isfinish=0;
	while(isfinish==0){
		int ret = handlerRequestIOFromWorker(sockfd);
		if (ret==ERR)
		{
			isfinish=1;
		}
	}
	printf("process4worker_thread has end\n");
	return NULL;
}
void* process4client_thread(void* arg){
	int sockfd=(int)arg;
	int isfinish=0;
	while(isfinish==0){
		int ret = handlerRequestIOFromClient(sockfd);
		if (ret==ERR)
		{
			isfinish=1;
		}
	}
	printf("process4client_thread has end\n");
	return NULL;
}

int handlerRequestIOFromClient(int clientID){
	Client2MasterRequest* client2MasterRequest = (Client2MasterRequest*)malloc(sizeof(Client2MasterRequest));
	int len = client2master_recv(client2MasterRequest,clientID);
	int resultNode=0;
	if (client2MasterRequest==NULL||len<=0)
	{
		printf("master recv the incorrect request from client len:%d\n",len);
		return ERR;
	}
	long requestID=client2MasterRequest->requestID;
	long long blockID=client2MasterRequest->blockID;
	int flag = client2MasterRequest->flag;
	free(client2MasterRequest);

	printf("master starts to hander requestID:%ld,blockID:%lld,flag:%d\n",requestID,blockID,flag);
	if (flag==0)
	{
		//执行read指令
		if (handlerReadRequestFromClient(blockID,requestID,clientID)==-1){
			printf("requestID:%ld handlerReadRequest is error\n", requestID);
			return ERR;
		}else{
			printf("requestID:%ld handlerReadRequest successed\n", requestID);
		}
	}else if (flag==1)
	{
		//执行write指令
		if (handlerWriteRequestFromClient(blockID,requestID,clientID)==-1){
			printf("requestID:%ld handlerWriteRequest is error\n", requestID);
			return ERR;
		}else{
			printf("requestID:%ld handlerWriteRequest successed\n", requestID);
		}
	}else{
		printf("requestID:%ld unkown flag:%d\n",requestID,flag);
		return ERR;
	}

	isBegin_archival_Record++;
	return 0;
}

void master2client_send_Resp4Read(Master2ClientResponse4Read* master2ClientResponse4Read,int clientID){
	size_t needSend=sizeof(Master2ClientResponse4Read);
	char *sendBuf=(char *)malloc(needSend);
	memcpy(sendBuf,master2ClientResponse4Read,needSend); //序列化
	send_bytes(client2master_sockfd[clientID],sendBuf,needSend); //master发送回复至client
	free(sendBuf);
}

void master2client_send_Resp4Write(Master2ClientResponse4Write* master2ClientResponse4Write,int clientID){
	size_t needSend=sizeof(Master2ClientResponse4Write);
	char *sendBuf=(char *)malloc(needSend);
	memcpy(sendBuf,master2ClientResponse4Write,needSend); //序列化
	send_bytes(client2master_sockfd[clientID],sendBuf,needSend); //master发送回复至client
	free(sendBuf);
}

int handlerWriteRequestFromClient(long long blockID,long requestID,int clientID){

	Master2ClientResponse4Write* master2ClientResponse4Write = (Master2ClientResponse4Write*)malloc(sizeof(Master2ClientResponse4Write));
	master2ClientResponse4Write->responseID=requestID;

	int primaryRackID,secondaryRackID,primaryNodeID,secondaryNodeID;
	pthread_rwlock_wrlock(&lock4SelectingRack);
	if (setCount==0)
	{
		//重新选择 rack
		currentRackID = selectRack();
		printf("reselect currentrackID%d\n",currentRackID);
	}

	primaryRackID=currentRackID;
	secondaryRackID=rack4SecondaryBlock[currentRackID];
	primaryNodeID=selectNode(primaryRackID);
	secondaryNodeID=selectNode(secondaryRackID);

	//包装response 发送给client
	master2ClientResponse4Write->nodeID1=primaryNodeID;
	master2ClientResponse4Write->nodeID2=secondaryNodeID;
	master2client_send_Resp4Write(master2ClientResponse4Write,clientID);
	free(master2ClientResponse4Write);
	printf("master2client_send has done\n");

	setCount = (setCount+1)%K;
	rack4SecondaryBlock[currentRackID] = (rack4SecondaryBlock[currentRackID]+1)%RackNumTotal;
	if (rack4SecondaryBlock[currentRackID]==currentRackID)
	{
		rack4SecondaryBlock[currentRackID] = (rack4SecondaryBlock[currentRackID]+1)%RackNumTotal;
	}
	pthread_rwlock_unlock(&lock4SelectingRack);
}

int handlerReadRequestFromClient(long long blockID,long requestID,int clientID){
	//加上读共享锁
	//pthread_rwlock_rdlock(&lock4BlockMeta);
	Master2ClientResponse4Read* master2ClientResponse4Read = (Master2ClientResponse4Read*)malloc(sizeof(Master2ClientResponse4Read));
	master2ClientResponse4Read->responseID=requestID;

	char* key = lltoa(blockID);
	BlockMeta* primary = (BlockMeta*)cuckoo_get(blockID2MetaMap,key);
	BlockMeta* secondary= (BlockMeta*)cuckoo_get(blockID2SecondMap,key);
	free(key);

	int selectNodeID;

	if (primary==NULL&&secondary==NULL)
	{
		printf("can not find the block blockID:%lld\n",blockID);
		master2ClientResponse4Read->nodeID1=-1;
		master2client_send_Resp4Read(master2ClientResponse4Read,clientID);
		free(master2ClientResponse4Read);
		printf("master2client_send has done\n");
		return 0;
	}
	//表明存在此块
	if (primary==NULL)
	{
		selectNodeID = secondary->nodeID;
	}else
	{
		if (primary->flag=='u')//为了保证读过程中，不会出现主副本归档删除的空档
		{
			selectNodeID = secondary->nodeID;
		}else
		{
			//随机副本访问
			int ran =randn(2);
			if (ran==1)
			{
				selectNodeID = primary->nodeID;
			}else
			{
				selectNodeID = secondary->nodeID;
			}
			
			// long prmaryReqNum4Rack = primary==NULL?-1:rackMetas[primary->rackID].requestNum;
			// long secondaryReqNum4Rack = secondary==NULL?-1:rackMetas[secondary->rackID].requestNum;
			// if (prmaryReqNum4Rack>secondaryReqNum4Rack)
			// {
			// 	selectNodeID = secondary->nodeID;
			// }else if (prmaryReqNum4Rack<secondaryReqNum4Rack)
			// {
			// 	selectNodeID = primary->nodeID;
			// }else
			// {
			// 	pthread_rwlock_rdlock(&lock4EnvMeta);
			// 	//如果rack的请求数相等 那么看node的请求数
			// 	long prmaryReqNum4Node = nodeMetas[primary->nodeID].requestNum;
			// 	long secondaryReqNum4Node = nodeMetas[secondary->nodeID].requestNum;
			// 	if (prmaryReqNum4Node==secondaryReqNum4Node)
			// 	{
			// 		int ran =randn(2);
			// 		if (ran==1)
			// 		{
			// 			selectNodeID = primary->nodeID;
			// 		}else
			// 		{
			// 			selectNodeID = secondary->nodeID;
			// 		}
			// 	}else
			// 	{
			// 		selectNodeID = prmaryReqNum4Node>secondaryReqNum4Node?secondary->nodeID:primary->nodeID;
			// 	}
			// 	pthread_rwlock_unlock(&lock4EnvMeta);
			// }
		}
	}

	//解开读共享锁

	//pthread_rwlock_unlock(&lock4BlockMeta);
	//包装response 发送给client
	master2ClientResponse4Read->nodeID1=selectNodeID;
	master2client_send_Resp4Read(master2ClientResponse4Read,clientID);
	free(master2ClientResponse4Read);
	printf("master2client_send has done\n");
	return 0;
}

int handlerRequestIOFromWorker(int worker2master_fd){
	Worker2MasterRequest* request = (Worker2MasterRequest*)malloc(sizeof(Worker2MasterRequest));
	int len = worker2master_recv(request,worker2master_fd);
	if (request==NULL||len<=0)
	{
		printf("master recv the incorrect request from worker len:%d\n",len);
		return ERR;
	}
	long requestID=request->requestID;
	int flag = request->flag;
	if (flag==1)
	{
		BlockMeta* blockMeta = (BlockMeta*)malloc(sizeof(BlockMeta));
		blockMeta->blockID=request->blockID;
		blockMeta->nodeID=request->nodeID;
		blockMeta->rackID=nodes[blockMeta->nodeID].rackID;
		blockMeta->NOA=0;
		blockMeta->isParity=0;
		blockMeta->RG_ID=-1;
		blockMeta->RG_Position=-1;
		blockMeta->timestamp=(long)time(NULL);
		//pthread_rwlock_wrlock(&lock4BlockMeta);
		char* key = lltoa(blockMeta->blockID);
		if (request->isPrimary==1)
		{
			blockMeta->flag='p';
			
			if(cuckoo_insert(blockID2MetaMap,key,(void*)blockMeta)==1) return ERR;
			pthread_rwlock_wrlock(&lock4BlockList);
			if(Array_List_Insert(blockList,(void*)blockMeta,-1)==-1) return ERR;
			pthread_rwlock_unlock(&lock4BlockList);
		}else{
			blockMeta->flag='s';
			if(cuckoo_insert(blockID2SecondMap,key,(void*)blockMeta)==1) return ERR;
		}

		//pthread_rwlock_unlock(&lock4BlockMeta);
		free(key);
		pthread_rwlock_wrlock(&lock4EnvMeta);
		rackMetas[blockMeta->rackID].blockNum++;
		nodeMetas[blockMeta->nodeID].blockNum++;
		int selectNodeID = request->nodeID;
		nodeMetas[selectNodeID].requestNum++;
		rackMetas[nodes[selectNodeID].rackID].requestNum++;
		pthread_rwlock_unlock(&lock4EnvMeta);

		printf("has update the metadata for writing block[%lld]\n",blockMeta->blockID);
	}else{
		//修改一些元数据包括：
		//1.数据块的元数据

		//pthread_rwlock_wrlock(&lock4BlockMeta);
		char* key = lltoa(request->blockID);
		BlockMeta* primary=(BlockMeta*)cuckoo_get(blockID2MetaMap,key);
		BlockMeta* secondary=(BlockMeta*)cuckoo_get(blockID2SecondMap,key);
		free(key);
		if (primary!=NULL) primary->NOA=primary->NOA+1;
		if (secondary!=NULL) secondary->NOA=secondary->NOA+1;
		//pthread_rwlock_unlock(&lock4BlockMeta);

		int selectNodeID = request->nodeID;
		pthread_rwlock_wrlock(&lock4EnvMeta);
		//2.nodemeta
		nodeMetas[selectNodeID].requestNum++;
		//3.rackmeta
		rackMetas[nodes[selectNodeID].rackID].requestNum++;
		pthread_rwlock_unlock(&lock4EnvMeta);
	}
	
	free(request);
	//发送requestID给worker
	send_bytes(worker2master_fd,(char*)&requestID,sizeof(long));
	printf("has sent the response to worker for finishing the work of updating\n");

	return 0;

}

int worker2master_recv(Worker2MasterRequest* request,int sockfd){
	size_t needRecv =sizeof(Worker2MasterRequest);
	char *recvBuf=(char *)malloc(needRecv*sizeof(char));
	int len = recv_bytes(sockfd,recvBuf,needRecv);
	memcpy(request,recvBuf,needRecv);  //反序列化
	return len;
}

int client2master_recv(Client2MasterRequest *request,int clientID){  //node 接收来自client的请求
	size_t needRecv =sizeof(Client2MasterRequest);
	char *recvBuf=(char *)malloc(needRecv*sizeof(char));
	int len = recv_bytes(client2master_sockfd[clientID],recvBuf,needRecv);
	memcpy(request,recvBuf,needRecv);  //反序列化
	return len;
}

//根据rack的上行负载选择 负载最小的rack作为放置rack
int selectRack(){
	int selectRackID = 0;
	int i;
	int len=0;
	int count[RackNumTotal]={0};
	for (i = 1; i < RackNumTotal; ++i)
	{
		if (rackMetas[i].requestNum<rackMetas[count[len]].requestNum)
		{
			len=0;
			count[0]=i;
		}

		if (rackMetas[i].requestNum==rackMetas[count[len]].requestNum)
		{
			len++;
			count[len]=i;
		}
	}
	int index = randn(len+1);
	selectRackID = count[index];
	return selectRackID;
}
//根据rackID，随机选择rack上的node
int selectNode(int rackID){
	int rnd = randn(racks[rackID].nodeNum);
	return racks[rackID].nodeIDs[rnd];
}

//从已有的数据块中筛选出不流行的数据块，加入到待归档数据块有序list中
int determineUnpopularBlocks(){
	printf("begin the function determineUnpopularBlocks\n");
	int i;
	long count=0;
	pthread_rwlock_wrlock(&lock4BlockList);
	for (i = 0; i <blockList->length-5; ++i)
	{
		count=count+((BlockMeta*)Array_List_GetAt(blockList,i))->NOA;
	}
	double averageFreq = count/(blockList->length-5);

	for (i = blockList->length-1-5; i >=0; i--)
	{
		if (averageFreq>((BlockMeta*)Array_List_GetAt(blockList,i))->NOA)
		{
			if (Sort_Way==0)
			{
				int index =randn(unpopularBlockList->length+1);
				Array_List_Insert(unpopularBlockList,(void*)Array_List_GetAt(blockList,i),index);		
			}else{
				Array_List_Insert(unpopularBlockList,(void*)Array_List_GetAt(blockList,i),0);
			}
			Array_List_RemoveAt(blockList,i);
			
		}
	}
	printf("end the function determineUnpopularBlocks and the len of unpopularBlockList is %ld\n",unpopularBlockList->length);
	pthread_rwlock_unlock(&lock4BlockList);
	fprintf(archival_file,"the len of unpopularBlockList is %ld\n",unpopularBlockList->length);
	return 0;
}

//排序flag表示时间局部性：1以及空间局部性：0
void sortByLocality(int flag){
	printf("begin the function sortByLocality\n");
	int i;
	if (flag==1)
	{
		printf("unpopularBlock sort by timestamp\n");
		//时间局部性
		//unpopularBlockList中的数据块已经按照时间排好，加入到不同racklist中，然后clear即可。
		for (i = 0; i <unpopularBlockList->length; ++i)
		{
			BlockMeta* blockMeta = (BlockMeta*)Array_List_GetAt(unpopularBlockList,i);
			Array_List_Insert(rackList[blockMeta->rackID],(void*)blockMeta,-1);
		}
	}else{
		printf("unpopularBlock sort by nodeID\n");
		//空间局部性
		//按照nodeID将unpopularBlockList中的数据块排好，加入到不同rackList中
		for (i = 0; i <unpopularBlockList->length; ++i)
		{
			BlockMeta* blockMeta = (BlockMeta*)Array_List_GetAt(unpopularBlockList,i);
			int j;
			for (j = rackList[blockMeta->rackID]->length-1; j>=0 ; j--)
			{
				BlockMeta* temp = (BlockMeta*)Array_List_GetAt(rackList[blockMeta->rackID],j);
				if (temp->nodeID<=blockMeta->nodeID)
				{
					break;
				}
			}
			Array_List_Insert(rackList[blockMeta->rackID],(void*)blockMeta,j+1);
		}
	}
	Array_List_Clear(unpopularBlockList);
	printf("end the function sortByLocality\n");
}


//在筛选出的不流行数据块有序list中，构建条带， 
int constructStripes(int flag){
	struct timeval stime,etime;
  	double time;

	printf("begin the function constructStripes using robinround\n");
	gettimeofday(&stime,NULL);

	determineUnpopularBlocks();

	gettimeofday(&etime,NULL);
	time=(etime.tv_sec-stime.tv_sec)+((double)(etime.tv_usec-stime.tv_usec)/1000000);
	fprintf(archival_file, "determineUnpopularBlocks_TIME:%lf\r\n",time);
	gettimeofday(&stime,NULL);

	sortByLocality(flag);

	gettimeofday(&etime,NULL);
	time=(etime.tv_sec-stime.tv_sec)+((double)(etime.tv_usec-stime.tv_usec)/1000000);
	fprintf(archival_file, "sortByLocality_TIME:%lf\r\n",time);
	gettimeofday(&stime,NULL);

	int i,j,k;
	//===========
	// archival_file=fopen("condition.txt","a+");
	// fprintf(archival_file, "after sorting===============================================================================\r\n");
	// for (i = 0; i < RackNumTotal; ++i)
	// {
	// 	for (j = 0; j < rackList[i]->length; ++j)
	// 	{
	// 		BlockMeta* meta = (BlockMeta*)Array_List_GetAt(rackList[i],j);
	// 		BlockMeta* meta2 = (BlockMeta*)cuckoo_get(blockID2SecondMap,lltoa(meta->blockID));
	// 		fprintf(archival_file, "BlockID:%lld,P-NodeID:%d,P-RackID:%d,S-NodeID:%d,S-RackID:%d\r\n",meta->blockID,meta->nodeID,meta->rackID,meta2->nodeID,meta2->rackID);
	// 	}
	// 	fprintf(archival_file, "===============================================================================\r\n");
	// }
	// fprintf(archival_file, "constructing stripes===============================================================================\r\n");
	//======================

	int stripeNum=0;
	int count=0;;	
	//在构成条带之后remove从racklist中
	//开始构建条带
	//对每个rack一次进行一次条带构建 直到没有满足限制的条带可以构建
	//条带构建算法：从rakclist中依次顺序选择数据块，如果满足条件继续选择，否则跳过选择下一个数据块，直到有K个数据块构成条带
	int isEndConstructing[RackNumTotal];//用来表示每个rack中是否还可以构建条带
	
	for (i = 0; i < RackNumTotal; ++i)
	{
		isEndConstructing[i]=0;
	}

	while(1){
		count=stripeNum;
		for (i = 0; i < RackNumTotal; ++i)
		{
			if (isEndConstructing[i]==1)//采用roundbin的方式 轮询每一个rack进行条带组织
			{
				continue;
			}else{
				int ret = constructStripesInRack(i);
				if (ret==1)
				{
					stripeNum++;//构建成功条带数加一
				}else{
					isEndConstructing[i]=1;//否则认为该rack内不存在符合限制的条带组织
				}
			}
		}
		if (count==stripeNum)//如果经过一次所有rack的轮询之后，没有条带数增加，认为当前集群已经条带组织完毕
		{
			break;
		}
	}

	//将rackList中还存在的块加入到blockList中，注意按照时间顺序
	pthread_rwlock_wrlock(&lock4BlockList);
	for (i = 0; i < RackNumTotal; ++i)
	{
		for (j = 0; j < rackList[i]->length; ++j)
		{
			//添加剩余块到blocklist中
			BlockMeta* restBlock = Array_List_GetAt(rackList[i],j);
			for (k = 0; k < blockList->length; ++k)
			{
				BlockMeta* tempBlock = Array_List_GetAt(blockList,k);
				if (restBlock->timestamp<tempBlock->timestamp)
				{
					Array_List_Insert(blockList,(void*)restBlock,k);
					break;
				}
			}
		}
		Array_List_Clear(rackList[i]);
	}
	pthread_rwlock_unlock(&lock4BlockList);
	printf("end the function constructStripes\n");
	//fclose(archival_file);
	gettimeofday(&etime,NULL);
	time=(etime.tv_sec-stime.tv_sec)+((double)(etime.tv_usec-stime.tv_usec)/1000000);
	fprintf(archival_file, "constructing_Stripe_TIME:%lf\r\n",time);

	return stripeNum;
}

//在单个rack内构建条带
int constructStripesInRack(int rackID){
	//printf("begin the function constructStripesInRack in rack[%d]\n",rackID);
	Array_List* list = rackList[rackID];
	int i,j,k,flag=0,blockNum=0;
	long blockIDs[K+R];
	int blockIndex[K];
	int secondaryRackIDs[K+R];
	for (i = K; i < K+R; ++i)
	{
		secondaryRackIDs[i]=-1;
	}


	for (i = 0; i < list->length; ++i)
	{
		BlockMeta* blockMeta = (BlockMeta*)Array_List_GetAt(list,i);
		blockIDs[blockNum]=blockMeta->blockID;
		char* key=lltoa(blockIDs[blockNum]);
		secondaryRackIDs[blockNum]=((BlockMeta*)cuckoo_get(blockID2SecondMap,key))->rackID;
		free(key);
		blockIndex[blockNum]=i;

		for ( k= 0; k < blockNum; k++)
		{
			if (secondaryRackIDs[k]==secondaryRackIDs[blockNum])//检查中，如果发现从副本的位置在条带中有重叠，表示不符合条件
			{
				flag=1;
				break;
			}
		}

		if (flag==0)
		{
			blockNum++;//当前块满足条件添加到条带中
		}
		flag=0;//flag重置为0
		if (blockNum==K)//表示条带已经够早完毕
		{
			break;
		}
	}


	// for (i = 0; i < list->length-K+1; i++)
	// {
	// 	blockNum=0;
	// 	for (j = i; j < list->length; j++)
	// 	{
	// 		//依次判断当前数据块位置是否符合限制
	// 		blockIDs[blockNum]=((BlockMeta*)Array_List_GetAt(list,j))->blockID;
	// 		char* key =lltoa(blockIDs[blockNum]) ;
	// 		secondaryRackIDs[blockNum]=((BlockMeta*)cuckoo_get(blockID2SecondMap,key))->rackID;
	// 		free(key);
	// 		blockIndex[blockNum]=j;

	// 		for ( k= 0; k < blockNum; k++)
	// 		{
	// 			if (secondaryRackIDs[k]==secondaryRackIDs[blockNum])//检查中，如果发现从副本的位置在条带中有重叠，表示不符合条件
	// 			{
	// 				flag=1;
	// 				break;
	// 			}
	// 		}

	// 		if (flag==0)
	// 		{
	// 			blockNum++;//当前块满足条件添加到条带中
	// 		}
	// 		flag=0;//flag重置为0
	// 		if (blockNum==K)//表示条带已经够早完毕
	// 		{
	// 			break;
	// 		}
	// 	}
	// 	if (blockNum==K)
	// 	{
	// 		break;//表示条带已经够早完毕
	// 	}
	// }

	//该rackList中没有符合限制的条带构建
	if (blockNum!=K)
	{
		printf("can not constructStripes in this rack,end the function constructStripesInRack in rack[%d]\n",rackID);
		return 0;
	}

	//printf("has sure the blocks in stripes next to malloc RG\n");
	//将blockIDs中的ID写到RG数据结构中，此外再选择R个rack作为校验块的存放位置

	RGroup* rgroup = (RGroup*)malloc(sizeof(RGroup));
	rgroup->RG_ID=RGID;
	rgroup->primaryRackID=rackID;
	for (i = K-1; i>=0; i--)
	{
		rgroup->blockIDs[i] = blockIDs[i];
		Array_List_RemoveAt(list,blockIndex[i]);

		//pthread_rwlock_wrlock(&lock4BlockMeta);
		char* key = lltoa(blockIDs[i]);
		BlockMeta* ogBlockMeta = (BlockMeta*)cuckoo_get(blockID2MetaMap,key);
		free(key);
		ogBlockMeta->flag='u';
		ogBlockMeta->RG_ID=RGID;
		ogBlockMeta->RG_Position=i;
		//pthread_rwlock_unlock(&lock4BlockMeta);
		//printf("add the %dth ogblock\n",i);
	}
	for (i = K; i < K+R; ++i)//为构成的条带中的校验数据块选择好位置
	{
		BlockMeta* paBlockMeta = (BlockMeta*)malloc(sizeof(BlockMeta));
		if (i==K)
		{
			paBlockMeta->rackID=rackID;
			paBlockMeta->nodeID=selectNode(rackID);
			secondaryRackIDs[i]=rackID;
		}else{
			while(1){
				int flag=0;
				int a = randn(RackNumTotal);
				for (j = 0; j < K+R; ++j)//bugs
				{
					if(a==secondaryRackIDs[j]) flag=1;
				}
				if (flag==1)
				{
					continue;//继续为校验数据块随机一个rackID,知道满足条带限制为止
				}else{
					secondaryRackIDs[i]=a;
					break;
				}
			}
			paBlockMeta->rackID=secondaryRackIDs[i];
			paBlockMeta->nodeID=selectNode(paBlockMeta->rackID);
		}

		while(1){
			long long rnd = (long long)rand();//随机一个校验数据块的ID
			char* key = lltoa(rnd);
			if (cuckoo_exists(parityID2MetaMap,key)==1)
			{
				free(key);
				continue;
			}else{
				free(key);
				paBlockMeta->blockID=rnd;
				break;
			}
		}

		paBlockMeta->isParity=1;
		paBlockMeta->RG_ID=RGID;
		paBlockMeta->RG_Position=i;
		char* key = lltoa(paBlockMeta->blockID);
		cuckoo_insert(parityID2MetaMap,key,(void*)paBlockMeta);
		free(key);
		rgroup->parityIDs[i-K]=paBlockMeta->blockID;
		//printf("add the %dth pablock parityBLockID:%lld\n",i,paBlockMeta->blockID);
		//printf("add the %dth pablock\n",i);
	}
	//todo 在构建条带后就删除primary的元信息，在stripeRequest中保留一份即可，但是不删除数据块
	//have done 当primary的元信息flag为’u‘时，表示待归档，读请求不会访问primary

	//RG加入到hashmap中，将RG_ID加入到一个list中表示未完成的stripe；
	rgroup->isFinish=0;
	char* key = ltoa(rgroup->RG_ID);
	cuckoo_insert(RGID2RGroupMap,key,(void*)rgroup);
	Array_List_Insert(ready2EncodingRGList,(void*)key,-1);
	RGID++;
	printf("success constructStripes in this rack,end the function constructStripesInRack in rack[%d]\n",rackID);
	

	//fprintf(archival_file, "BlockID1:%lld,BlockID2:%lld,BlockID3:%lld,BlockID4:%lld\r\n",rgroup->blockIDs[0],rgroup->blockIDs[1],rgroup->blockIDs[2],rgroup->blockIDs[3]);

	return 1;
}

int isArchived=0;
//归档的线程函数
void* archival_thread(void* arg){
	// 每隔一段时间开启归档进程
	while(1){
		if (isBegin_archival_Record==Archival_Time)
		{
			archival_file = fopen("archival.txt","a+");
			processArchival();
			fclose(archival_file);
			isArchived=1;		
			return NULL;
		}
	}
}

//归档的线程函数
/*
1.Loop ready2EncodingRGList中每一个RG_ID
2.获取RGroup
3.创建一个归档request
4.发送request给已经指定的编码节点
5.recv归档response来自编码节点
6.删除或者修改相应的metadata
7.free空间资源
*/
void processArchival(){

	struct timeval stime,etime;
  	double time;

	printf("The system has started the work of archiving \n");
	int stripeNum = constructStripes(Sort_Way);//数据块筛选并且进行条带组织
	fprintf(archival_file,"%d stripes need to be archived\n", stripeNum);

	gettimeofday(&stime,NULL);

	int i,j;
	//1.Loop ready2EncodingRGList中每一个RG_ID
	RGroup* rgroup=NULL;
	char* key=NULL;
	BlockMeta* primaryMeta[K];
	BlockMeta* parityMeta[R];
	ArchivalRequest* archivalRequest = (ArchivalRequest*)malloc(sizeof(ArchivalRequest));
	for (i = 0; i < ready2EncodingRGList->length; ++i)
	{
		//2.获取RGroup
		key = (char*)Array_List_GetAt(ready2EncodingRGList,i);
		rgroup = (RGroup*)cuckoo_get(RGID2RGroupMap,key);
		if (rgroup->isFinish==1){
			printf("RG_ID[%ld] the work of archive has been finish\n",rgroup->RG_ID);
			continue;
		}
		//3.创建一个归档request并赋值
		printf("create a archival request\n");
		archivalRequest->RG_ID=rgroup->RG_ID;

		//pthread_rwlock_rdlock(&lock4BlockMeta);
		//pthread_rwlock_wrlock(&lock4EnvMeta);
		for (j = 0; j < K; ++j){
			//设置源数据块的id
			archivalRequest->blockIDs[j] = rgroup->blockIDs[j];
			//设置源数据块的主副本的nodeID
			char* key = lltoa(rgroup->blockIDs[j]);
			primaryMeta[j] = (BlockMeta*)cuckoo_get(blockID2MetaMap,key);
			free(key);
			archivalRequest->originalNodeIDs[j]=primaryMeta[j]->nodeID;
		}
		//pthread_rwlock_unlock(&lock4BlockMeta);
		for (j = 0; j < R; ++j){
			//设置校验数据块的id
			archivalRequest->parityIDs[j] = rgroup->parityIDs[j];
			//设置校验数据块的目标位置的nodeID
			char* key =lltoa(rgroup->parityIDs[j]) ;
			parityMeta[j] = (BlockMeta*)cuckoo_get(parityID2MetaMap,key);
			free(key);
			archivalRequest->parityNodeIDs[j]=parityMeta[j]->nodeID;
		}
		//4.发送request给已经指定的编码节点
		printf("send a archival request and select a node in Rack[%d]\n",rgroup->primaryRackID);
		int encodingNodeID = selectEncodingNode(archivalRequest,rgroup->primaryRackID);
		parityMeta[0]->nodeID=encodingNodeID;
		archivalRequest->parityNodeIDs[0]=encodingNodeID;
		
		master2worker_send(archivalRequest,encodingNodeID);
		
		printf("archivalRequest of RG_ID[%ld] has been sent to nodeID[%d]\n", archivalRequest->RG_ID,encodingNodeID);

		//5.recv归档response来自编码节点
		long recvRG_ID;
		recv_bytes(masterAsClient2WorkerConnect[encodingNodeID].master2worker_sockfd,(char*)&recvRG_ID,sizeof(long));
		
		printf("recv the response from worker[%d]\n",encodingNodeID);
		if (recvRG_ID==rgroup->RG_ID)
		{
			for (j = 0; j < K; ++j){
				//1.删除原始数据块的meta
				char* key =lltoa(rgroup->blockIDs[j]);
				BlockMeta* temp  = (BlockMeta*)cuckoo_remove(blockID2MetaMap,key);
				nodeMetas[temp->nodeID].blockNum--;
				rackMetas[temp->rackID].blockNum--;//更新rack以及node中数据块的个数

				// if (temp->nodeID!=encodingNodeID)
				// {
				// 	nodeMetas[temp->nodeID].requestNum=nodeMetas[temp->nodeID].requestNum+2;
				// }

				// if (nodes[temp->nodeID].rackID!=nodes[encodingNodeID].rackID)
				// {
				// 	rackMetas[nodes[temp->nodeID].rackID].requestNum=rackMetas[nodes[temp->nodeID].rackID].requestNum+2;
				// }

				if (temp!=NULL) free(temp);
				//2.修改从副本的meta
				BlockMeta* secondaryMeta = (BlockMeta*)cuckoo_get(blockID2SecondMap,key);
				secondaryMeta->flag='e';
				secondaryMeta->RG_ID=rgroup->RG_ID;
				secondaryMeta->RG_Position=j;
				free(key);
			}

			for (j = 0; j < R; ++j){
				//3.更新校验块所在node以及rack的数据块个数
				nodeMetas[parityMeta[j]->nodeID].blockNum++;
				rackMetas[parityMeta[j]->rackID].blockNum++;

				// if (parityMeta[j]->nodeID!=encodingNodeID)
				// {
				// 	nodeMetas[parityMeta[j]->nodeID].requestNum++;
				// }

				// if (nodes[parityMeta[j]->nodeID].rackID!=nodes[encodingNodeID].rackID)
				// {
				// 	rackMetas[nodes[parityMeta[j]->nodeID].rackID].requestNum++;
				// }
			}
			rgroup->isFinish=1;
			printf("RG_ID[%ld] has been archived successfully\n",recvRG_ID);

		}else{
			for (j = 0; j < K; ++j){
				//1.还原数据块的meta
				primaryMeta[j]->flag='p';
				primaryMeta[j]->RG_ID=-1;
				primaryMeta[j]->RG_Position=-1;
			}
			//2.删除parity的meta
			for (j = 0; j < R; ++j) {
				char* key = lltoa(rgroup->parityIDs[j]);
				BlockMeta* temp  = (BlockMeta*)cuckoo_remove(parityID2MetaMap,key);
				if (temp!=NULL) free(temp);
				free(key);
			}
			//3.删除rgroup
			char* key = ltoa(rgroup->RG_ID);
			cuckoo_remove(RGID2RGroupMap,key);
			free(key);
			if (rgroup!=NULL) free(rgroup);
			printf("RG_ID[%ld] is failed to be archived \n",recvRG_ID);
		}
		// pthread_rwlock_unlock(&lock4EnvMeta);
		// pthread_rwlock_unlock(&lock4BlockMeta);

	}
	//7.free空间资源
	if (archivalRequest!=NULL) free(archivalRequest);
	Array_List_Clear(ready2EncodingRGList);	
	printf("The system has finished the work of archiving \n");

	gettimeofday(&etime,NULL);
	time=(etime.tv_sec-stime.tv_sec)+((double)(etime.tv_usec-stime.tv_usec)/1000000);
	fprintf(archival_file, "archiving_TIME:%lf\r\n",time);
}

//为一个条带选择编码节点，编码节点中应该含有最多的原始数据块数
int selectEncodingNode(ArchivalRequest* archivalRequest,int rackID){
	int nodeNum = racks[rackID].nodeNum;
	int* count = (int*)malloc(sizeof(int)*nodeNum);
	int i,j,ret=randn(nodeNum);
	for (i = 0; i < nodeNum; ++i) count[i]=0;
	for (i = 0; i < K; ++i)
	{
		for (int j = 0; j < nodeNum; ++j)//一次遍历，对所在nodeid的count加1
		{
			if (racks[rackID].nodeIDs[j]==archivalRequest->originalNodeIDs[i])
			{
				count[j]++;
				if(count[j]>count[ret]) ret=j;
				break;
			}
		}
	}
	if (count!=NULL) free(count);
	return racks[rackID].nodeIDs[ret];
}

//master向worker发送归档请求
void master2worker_send(ArchivalRequest *request,int nodeID){
	size_t needSend=sizeof(ArchivalRequest);
	char *sendBuf=(char *)malloc(needSend);
	memcpy(sendBuf,request,needSend); //序列化
	send_bytes(masterAsClient2WorkerConnect[nodeID].master2worker_sockfd,sendBuf,needSend); //Client发送请求至worker
	if (sendBuf!=NULL) free(sendBuf);
}

void printCondition(){
	load_file = fopen("load.txt","a+");
	for (int i = 0; i < NodeNumTotal-1; ++i)
	{
		fprintf(load_file,"nodeID:%d,requestNum:%ld,blockNum:%ld\r\n",nodeMetas[i].nodeID,nodeMetas[i].requestNum,nodeMetas[i].blockNum);
	}
	for (int i = 0; i < RackNumTotal; ++i)
	{
		fprintf(load_file,"rackID:%d,requestNum:%ld,blockNum:%ld\r\n",rackMetas[i].rackID,rackMetas[i].requestNum,rackMetas[i].blockNum);
	}
	fprintf(load_file, "===============================================================================\r\n");
	fclose(load_file);
}

void eviction(){
	Array_List* NOAlist = Array_List_Init();
	for (int i = 0; i < blockID2SecondMap->cap; i++){
	 	if (blockID2SecondMap->nodes[i].taken==1){
	 		BlockMeta* blockMeta2=(BlockMeta*)blockID2SecondMap->nodes[i].val;
	 		int j=0;
	 		for (j = 0; j < NOAlist->length; ++j)
	 		{
	 			BlockMeta* blockMeta1=(BlockMeta*)Array_List_GetAt(NOAlist,j);
	 			if (blockMeta2->NOA<blockMeta1->NOA)
	 			{
	 				break;
	 			}
	 		}
	 		Array_List_Insert(NOAlist,(void*)blockMeta2,j);
	 	}
	}

	//Eviction_Proption
	int count = NOAlist->length*Eviction_Proption/100;

	int* rgEvictionCount=(int*)malloc(sizeof(int)*RGID);
	for (int i = 0; i < RGID; ++i) rgEvictionCount[i]=0;

	for (int i = 0; i < count; ++i)
	{
		BlockMeta* temp = (BlockMeta*)Array_List_GetAt(NOAlist,i);
		if (temp->RG_ID!=-1)
		{
			rgEvictionCount[temp->RG_ID]++;
		}
	}

	int res=0;
	for (int i = 0; i < RGID; ++i){
		//printf("Eviction stage: RG_ID:%d,evicted blocknum:%d\n",i,rgEvictionCount[i]);
		if (rgEvictionCount[i]==K) res++;
	}

	printf("count:%d\n",count);
	printf("Eviction stage:%d stripes has been evicted in the end\n",res);
	Array_List_Free(NOAlist);
	free(rgEvictionCount);
}

int main()
{
	int ret;
	ret = init_master();
	if (ret==ERR){
		printf("fail to init master\n");
		destory_master();
		return 0;
	}

	//为client创建线程
	pthread_t thread4clients[ClientNum];
	for (int i = 0; i < ClientNum; ++i)
	{
		pthread_create(&thread4clients[i],NULL,process4client_thread,(void*)i);
	}
	//为worker创建线程
	pthread_t thread4workers[NodeNumTotal-1][ClientNum];

	for (int i = 0; i < NodeNumTotal-1; ++i)
	{
		for (int j = 0; j < ClientNum; ++j)
		{
			pthread_create(&thread4workers[i][j],NULL,process4worker_thread,(void*)nodeMetas[i].worker2master_sockfd[j]);
		}		
	}

	//创建归档线程
	//todo
	if (IsBegin_Archival_Process==1)
	{
		pthread_t archival_threadID;
		pthread_create(&archival_threadID,NULL,archival_thread,NULL);
	}
	

	while(1){
		sleep(1);
		if (isArchived==1||IsBegin_Archival_Process==0) break;
	}


	while(1){
		if (isBegin_archival_Record==100001)
		{
			sleep(1);
			printCondition();
			isDestory=1;
			//添加evcition阶段
			eviction();
		}
		if(isDestory==1){
			break;
		}
	}

	for (int i = 0; i < ClientNum; ++i)
	{
		pthread_cancel(thread4clients[i]);
	}

	for (int i = 0; i < NodeNumTotal-1; ++i)
	{
		for (int j = 0; j < ClientNum; ++j)
		{
			pthread_cancel(thread4workers[i][j]);
		}
		
	}


	destory_master();
	return 0;
}